package com.yl.flink.source;

import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import com.yl.util.FlinkUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;

@Slf4j
public class TDengineSource extends RichSourceFunction<MultiDataEntity> {

    private final String jdbcUrl;
    private transient Statement statement;
    private transient Connection connection;

    public TDengineSource(ParameterTool params) {
        this.jdbcUrl = params.get(Const.TDENGINE_URL);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = FlinkUtil.getTDengineConn(this.jdbcUrl);
        statement = connection.createStatement();
    }

    @Override
    public void run(SourceContext<MultiDataEntity> sourceContext) throws Exception {
        try {
            String sql = "select * from db_common.t_1";
            ResultSet resultSet = statement.executeQuery(sql);
            while (resultSet.next()) {
                long timestamp = resultSet.getLong(1);
                MultiDataEntity multiDataEntity = MultiDataEntity.builder().build();
                sourceContext.collect(multiDataEntity);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
        if (statement != null) {
            statement.close();
        }
    }

    @Override
    public void cancel() {
    }
}
